package com.ecommerce.recommendation.util;

import net.minidev.json.JSONObject;
import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.CSVParser;
import org.apache.commons.csv.CSVRecord;
import org.apache.commons.math3.stat.regression.SimpleRegression;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

@Component
public class MapReduceUtil {
    static {
        // 重要：指向 Hadoop 根目录（含 bin 子目录）
        System.setProperty("hadoop.home.dir", "E:\\hadoop-3.2.0");
    }

    // 虚拟机Hadoop配置（需根据实际情况修改）
    private static final String HDFS_URL = "hdfs://192.168.74.100:9000"; // 虚拟机HDFS地址
    // 根据实际文件路径修改
    private static final String CUSTOMER_DATA_PATH = "/user/hadoop/ecommerce/input/customer_data_collection.csv";
    private static final String PRODUCT_RECOMMENDATION_DATA_PATH = "/user/hadoop/ecommerce/input/product_recommendation_data.csv";

    // 新增：创建Hadoop配置方法
    private Configuration createHadoopConfiguration() {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", HDFS_URL); // 配置HDFS默认地址
        return conf;
    }

    // 顾客行为分析（添加详细日志）
    public void runCustomerBehaviorAnalysis() {
        try {
            Configuration conf = createHadoopConfiguration();
            Job job = Job.getInstance(conf, "CustomerBehaviorAnalysis");
            job.setJarByClass(MapReduceUtil.class);

            // 设置Mapper和Reducer类
            job.setMapperClass(CustomerBehaviorMapper.class);
            job.setReducerClass(CustomerBehaviorReducer.class);

            // 设置Mapper输出的键值类型
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(Text.class);

            // 设置最终输出的键值类型
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(Text.class);

            // 设置输入输出路径
            FileInputFormat.addInputPath(job, new Path(CUSTOMER_DATA_PATH));
            FileOutputFormat.setOutputPath(job, new Path(HDFS_URL + "/user/hadoop/ecommerce/output/customer-behavior"));

            // 执行作业
            boolean success = job.waitForCompletion(true);
            if (!success) {
                throw new RuntimeException("顾客行为分析作业执行失败");
            }
        } catch (Exception e) {
            throw new RuntimeException("顾客行为分析失败: " + e.getMessage(), e);
        }
    }

    // 产品推荐分析
    public void runProductRecommendationAnalysis() {
        try {
            Configuration conf = createHadoopConfiguration();
            Job job = Job.getInstance(conf, "ProductRecommendationAnalysis");
            job.setJarByClass(MapReduceUtil.class);

            job.setMapperClass(ProductRecommendationMapper.class);
            job.setReducerClass(ProductRecommendationReducer.class);

            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(Text.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(DoubleWritable.class);

            FileInputFormat.addInputPath(job, new Path(PRODUCT_RECOMMENDATION_DATA_PATH));
            FileOutputFormat.setOutputPath(job, new Path(HDFS_URL + "/user/hadoop/ecommerce/output/product-recommendation"));

            if (!job.waitForCompletion(true)) {
                throw new RuntimeException("ProductRecommendationAnalysis job failed");
            }
        } catch (IOException | ClassNotFoundException | InterruptedException e) {
            throw new RuntimeException("Product recommendation analysis failed", e);
        }
    }

    // 季节性趋势分析
    public void runSeasonalTrendAnalysis() {
        try {
            Configuration conf = createHadoopConfiguration();
            Job job = Job.getInstance(conf, "SeasonalTrendAnalysis");
            job.setJarByClass(MapReduceUtil.class);

            job.setMapperClass(SeasonalTrendMapper.class);
            job.setReducerClass(SeasonalTrendReducer.class);

            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(IntWritable.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);

            FileInputFormat.addInputPath(job, new Path(CUSTOMER_DATA_PATH));
            FileOutputFormat.setOutputPath(job, new Path(HDFS_URL + "/user/hadoop/ecommerce/output/seasonal-trend"));

            if (!job.waitForCompletion(true)) {
                throw new RuntimeException("SeasonalTrendAnalysis job failed");
            }
        } catch (IOException | ClassNotFoundException | InterruptedException e) {
            throw new RuntimeException("Seasonal trend analysis failed", e);
        }
    }

    // 区域市场分析
    public void runRegionalMarketAnalysis() {
        try {
            Configuration conf = createHadoopConfiguration();
            Job job = Job.getInstance(conf, "RegionalMarketAnalysis");
            job.setJarByClass(MapReduceUtil.class);

            // Mapper 和 Reducer 配置
            job.setMapperClass(RegionalMarketMapper.class);
            job.setReducerClass(RegionalMarketReducer.class);

            // 输出类型配置
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(Text.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(Text.class); // 假设输出为文本格式（城市,热门品类,销量）

            // 输入输出路径配置
            FileInputFormat.addInputPath(job, new Path(PRODUCT_RECOMMENDATION_DATA_PATH));
            // 输出到 HDFS 文本路径（注意：需提前确保路径不存在或可覆盖）
            Path outputPath = new Path(HDFS_URL + "/user/hadoop/ecommerce/output/regional-market");
            FileOutputFormat.setOutputPath(job, outputPath);

            if (!job.waitForCompletion(true)) {
                throw new RuntimeException("RegionalMarketAnalysis job failed");
            }
        } catch (IOException | ClassNotFoundException | InterruptedException e) {
            throw new RuntimeException("Regional market analysis failed", e);
        }
    }


    // 用户留存分析
    public void runUserRetentionAnalysis() {
        try {
            Configuration conf = createHadoopConfiguration();
            Job job = Job.getInstance(conf, "UserRetentionAnalysis");
            job.setJarByClass(MapReduceUtil.class);

            job.setMapperClass(UserRetentionMapper.class);
            job.setReducerClass(UserRetentionReducer.class);

            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(Text.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(Text.class);

            FileInputFormat.addInputPath(job, new Path(CUSTOMER_DATA_PATH));
            FileOutputFormat.setOutputPath(job, new Path(HDFS_URL + "/user/hadoop/ecommerce/output/user-retention"));

            if (!job.waitForCompletion(true)) {
                throw new RuntimeException("UserRetentionAnalysis job failed");
            }
        } catch (IOException | ClassNotFoundException | InterruptedException e) {
            throw new RuntimeException("User retention analysis failed", e);
        }
    }

    // ====================================
    // Mapper 和 Reducer 具体实现
    // ====================================

    // 顾客行为分析 Mapper
// 字段索引常量
    private static final int AGE_INDEX = 1;
    private static final int GENDER_INDEX = 2;
    private static final int CITY_INDEX = 3;
    private static final int CATEGORY_INDEX = 4;
    private static final int AMOUNT_INDEX = 7;

    public static class CustomerBehaviorMapper extends Mapper<LongWritable, Text, Text, Text> {
        private final Text outputKey = new Text();
        private final Text outputValue = new Text();
        private boolean isHeader = true;

        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            String line = value.toString().trim();

            // 标题行处理
            if (isHeader) {
                if (line.startsWith("Customer_ID,Age,Gender")) {
                    isHeader = false;
                    System.out.println("跳过标题行: " + line);
                    return;
                } else {
                    System.err.println("警告: 非预期标题行: " + line);
                    isHeader = false;
                }
            }

            try {
                // 使用 commons-csv 解析 CSV
                CSVParser parser = CSVParser.parse(line, CSVFormat.DEFAULT.withIgnoreSurroundingSpaces());
                List<CSVRecord> records = new ArrayList<>(parser.getRecords());
                if (records.isEmpty() || records.get(0).size() < 10) {
                    System.err.println("字段不足，跳过: " + line);
                    return;
                }
                CSVRecord record = records.get(0);

                // 年龄解析
                String ageStr = record.get(AGE_INDEX);
                if (!ageStr.matches("\\d+")) {
                    System.err.println("无效年龄: " + ageStr + "，行: " + line);
                    return;
                }
                int age = Integer.parseInt(ageStr);

                // 基础字段
                String gender = record.get(GENDER_INDEX);
                String city = record.get(CITY_INDEX);

                // 类别清洗（新增：移除多余撇号）
                String categoryRaw = record.get(CATEGORY_INDEX);
                String categoryClean = categoryRaw.replaceAll("^\\['|'\\]$|^\\[|\\]$", "");
                String[] categories = categoryClean.split(", ");
                if (categories.length == 0) {
                    System.err.println("无效类别: " + categoryRaw + "，行: " + line);
                    return;
                }
                // 新增：标准化品类名称，移除多余撇号
                String category = categories[0].replaceAll("'$", "");

                // 金额解析
                String amountStr = record.get(AMOUNT_INDEX);
                if (!amountStr.matches("\\d+(\\.\\d+)?|\\d+\\.\\d+[eE]\\d+")) {
                    System.err.println("无效金额: " + amountStr + "，行: " + line);
                    return;
                }
                double amount = Double.parseDouble(amountStr.replace("元", ""));

                // 输出
                String ageGroup = getAgeGroup(age);
                outputKey.set(ageGroup + "," + gender + "," + city);
                outputValue.set(category + ":" + amount);
                context.write(outputKey, outputValue);

            } catch (NumberFormatException e) {
                System.err.println("数值错误: " + e.getMessage() + "，行: " + line);
            } catch (Exception e) {
                System.err.println("未知错误: " + e.getMessage() + "，行: " + line);
            }
        }

        private String getAgeGroup(int age) {
            if (age < 20) return "0-19";
            if (age < 30) return "20-29";
            if (age < 40) return "30-39";
            if (age < 50) return "40-49";
            return "50+";
        }
    }

    public static class CustomerBehaviorReducer extends Reducer<Text, Text, Text, Text> {
        private final Text result = new Text();

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            Map<String, Double> categoryTotal = new HashMap<>(16);
            Map<String, Integer> categoryCount = new HashMap<>(16);

            for (Text value : values) {
                String[] parts = value.toString().split(":");
                if (parts.length != 2) continue;

                String category = parts[0];
                double amount = Double.parseDouble(parts[1]);

                categoryTotal.merge(category, amount, Double::sum);
                categoryCount.merge(category, 1, Integer::sum);
            }

            // 构建结果
            StringBuilder sb = new StringBuilder();
            for (Map.Entry<String, Integer> entry : categoryCount.entrySet()) {
                String category = entry.getKey();
                int count = entry.getValue();
                double total = categoryTotal.get(category);
                double avg = total / count;

                sb.append(String.format("%s(销量:%d, 金额:%.2f元, 均价:%.2f元), ",
                        category, count, total, avg));
            }

            // 总消费计算
            double totalConsumption = categoryTotal.values().stream()
                    .mapToDouble(Double::doubleValue).sum();

            // 收尾处理
            if (sb.length() > 0) {
                sb.setLength(sb.length() - 2); // 移除最后 ", "
            }
            sb.append(" | 总消费: ").append(String.format("%.2f", totalConsumption)).append("元");

            result.set(sb.toString());
            context.write(key, result);
        }
    }

    // 产品推荐分析 Mapper
    public static class ProductRecommendationMapper extends Mapper<LongWritable, Text, Text, Text> {
        private final Text outputKey = new Text();
        private final Text outputValue = new Text();
        private boolean isHeader = true;
        private List<String> headerNames = new ArrayList<>();
        private int expectedColumnCount = -1;
        private static final String DEFAULT_KEY = "GLOBAL"; // 默认全局聚合 Key

        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            String line = value.toString().trim();
            if (line.isEmpty()) return;

            try {
                if (isHeader) {
                    // 解析标题行
                    CSVFormat headerFormat = CSVFormat.DEFAULT
                            .withQuote('"')
                            .withIgnoreSurroundingSpaces()
                            .withTrim()
                            .withHeader();

                    CSVParser headerParser = CSVParser.parse(line, headerFormat);
                    headerNames = Arrays.asList(headerParser.getHeaderMap().keySet().toArray(new String[0]));
                    expectedColumnCount = headerNames.size();
                    isHeader = false;
                    System.out.println("标题行列数: " + expectedColumnCount + ", 列名: " + headerNames);
                    return;
                }

                // 解析数据行
                CSVFormat dataFormat = CSVFormat.DEFAULT
                        .withQuote('"')
                        .withIgnoreSurroundingSpaces()
                        .withTrim()
                        .withSkipHeaderRecord(true);

                CSVParser parser = CSVParser.parse(line, dataFormat);
                List<CSVRecord> records = parser.getRecords();
                if (records.isEmpty()) {
                    System.err.println("无有效记录，跳过: " + line);
                    return;
                }

                CSVRecord record = records.get(0);
                List<String> fieldValues = new ArrayList<>(expectedColumnCount);

                // 填充字段值，处理列数不一致问题
                for (int i = 0; i < expectedColumnCount; i++) {
                    fieldValues.add(i < record.size() ? record.get(i).trim() : "");
                }

                // 字段索引（可根据实际数据调整）
                int priceIndex = headerNames.indexOf("Price");
                int productRatingIndex = headerNames.indexOf("Average_Rating_of_Similar_Products");
                int sentimentRatingIndex = headerNames.indexOf("Customer_Review_Sentiment_Score");
                int isRecommendedIndex = headerNames.indexOf("Holiday"); // 假设 Holiday 列表示是否推荐
                int brandIndex = headerNames.indexOf("Brand");
                int similarProductListIndex = headerNames.indexOf("Similar_Product_List");

                // 提取字段（增加空值处理）
                double price = parseDouble(fieldValues.get(priceIndex));
                double productRating = parseDouble(fieldValues.get(productRatingIndex));
                double sentimentRating = parseDouble(fieldValues.get(sentimentRatingIndex));
                boolean isRecommended = "Yes".equalsIgnoreCase(fieldValues.get(isRecommendedIndex));
                String brand = fieldValues.get(brandIndex).toUpperCase().trim(); // 统一品牌大小写
                List<String> similarProductList = parseProductList(fieldValues.get(similarProductListIndex));

                // 关键修改：设置 Key 为品牌（或 DEFAULT_KEY 进行全局聚合）
                outputKey.set(brand.isEmpty() ? DEFAULT_KEY : brand); // 空品牌归为全局

                // 构造输出值（相似产品用分号分隔，与 Reducer 解析一致）
                String similarProductStr = similarProductList.stream()
                        .filter(s -> !s.isEmpty())
                        .collect(Collectors.joining(";")); // 使用分号分隔

                String outputStr = String.format("%f,%f,%f,%b,%s,%s",
                        price, productRating, sentimentRating, isRecommended, brand, similarProductStr);
                outputValue.set(outputStr);
                context.write(outputKey, outputValue);

            } catch (Exception e) {
                System.err.println("解析错误: " + e.getMessage() + ", 行: " + line);
            }
        }

        // 安全解析 double（增加默认值处理）
        private double parseDouble(String value) {
            if (value.isEmpty() || "null".equalsIgnoreCase(value)) return 0;
            try {
                return Double.parseDouble(value);
            } catch (NumberFormatException e) {
                System.err.println("数字解析错误: " + value);
                return 0;
            }
        }

        // 解析产品列表（支持空列表和标准格式）
        private List<String> parseProductList(String listStr) {
            if (listStr == null || listStr.isEmpty()) return Collections.emptyList();
            // 处理两种格式：[P1,P2] 或 P1;P2（根据实际数据调整）
            if (listStr.startsWith("[") && listStr.endsWith("]")) {
                return Arrays.stream(listStr.substring(1, listStr.length() - 1).split(","))
                        .map(String::trim)
                        .filter(s -> !s.isEmpty())
                        .collect(Collectors.toList());
            } else {
                return Arrays.stream(listStr.split(";"))
                        .map(String::trim)
                        .filter(s -> !s.isEmpty())
                        .collect(Collectors.toList());
            }
        }
    }


    // Reducer 改造：简单线性回归 + 协同过滤（按品牌分组版）
    public static class ProductRecommendationReducer
            extends Reducer<Text, Text, Text, Text> {

        private static final int MIN_REQUIRED_SAMPLES = 5;

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {

            List<Double> yList = new ArrayList<>();
            List<Double> xList = new ArrayList<>();
            List<List<String>> allSimilarLists = new ArrayList<>();

            // 数据解析（保持不变）
            for (Text value : values) {
                String[] parts = value.toString().split(",");
                if (parts.length < 6) continue;
                double price = Double.parseDouble(parts[0]);
                boolean isRecommended = Boolean.parseBoolean(parts[3]);
                List<String> similarList = Arrays.stream(parts[5].split(";"))
                        .map(s -> s.replace("'", "")) // 移除单引号
                        .filter(s -> !s.isEmpty())
                        .collect(Collectors.toList());
                xList.add(price);
                yList.add(isRecommended ? 1.0 : 0.0);
                allSimilarLists.add(similarList);
            }

            int sampleSize = xList.size();
            boolean hasEnoughData = sampleSize >= MIN_REQUIRED_SAMPLES;
            List<String> topSimilar = new ArrayList<>();
            double priceWeight = 0.0; // 关键因子权重（Price）
            double regressionScore = 0.0;
            String warning = "";

            // 协同过滤（保持不变）
            Map<String, Integer> productFreq = new HashMap<>();
            for (List<String> list : allSimilarLists) {
                for (String product : list) {
                    productFreq.put(product, productFreq.getOrDefault(product, 0) + 1);
                }
            }
            topSimilar = productFreq.entrySet().stream()
                    .filter(entry -> !entry.getKey().equals(key.toString()))
                    .sorted(Map.Entry.comparingByValue(Comparator.reverseOrder()))
                    .limit(5)
                    .map(Map.Entry::getKey)
                    .collect(Collectors.toList());

            // 回归分析（提取权重并格式化）
            if (hasEnoughData) {
                try {
                    SimpleRegression regression = new SimpleRegression();
                    for (int i = 0; i < sampleSize; i++) {
                        regression.addData(xList.get(i), yList.get(i));
                    }
                    priceWeight = regression.getSlope(); // 直接获取权重值
                    regressionScore = regression.getRSquare();
                } catch (Exception e) {
                    warning = "回归分析失败: " + e.getMessage();
                }
            } else {
                warning = String.format("样本量不足 (%d条，需≥%d条)", sampleSize, MIN_REQUIRED_SAMPLES);
            }

            // 构造JSON对象
            JSONObject result = new JSONObject();
            result.put("品牌", key.toString());
            result.put("回归得分", String.format("%.8f", regressionScore));
            result.put("关键因子权重", Collections.singletonMap("Price", String.format("%.10f", priceWeight)));
            result.put("相似产品", topSimilar);
            result.put("警告", warning.isEmpty() ? "无" : warning);

            // 输出完整JSON字符串
            context.write(key, new Text(result.toJSONString()));
        }
    }

    // 季节性趋势分析 Mapper，需根据实际 CSV 列顺序调整索引
    public static class SeasonalTrendMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
        private final Text outputKey = new Text();
        private final IntWritable one = new IntWritable(1);
        private boolean isHeader = true;
        // 字段索引，根据实际数据列顺序调整
        private static final int SEASON_INDEX = 9;   // Season 是第10列，索引为9
        private static final int HOLIDAY_INDEX = 8;  // Holiday 是第9列，索引为8
        private static final int CATEGORY_INDEX = 5; // Purchase_History 是第6列，索引为5

        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            String line = value.toString().trim();

            // 标题行处理
            if (isHeader) {
                if (line.startsWith("Customer")) { // 根据实际标题行的第一个字段调整
                    isHeader = false;
                    System.out.println("跳过标题行: " + line);
                    return;
                } else {
                    System.err.println("警告: 非预期标题行: " + line);
                    // 继续处理，假设这不是标题行
                }
            }

            try {
                // 使用 commons-csv 解析 CSV，禁用 withHeader()
                CSVParser parser = CSVParser.parse(line, CSVFormat.DEFAULT
                        .withIgnoreSurroundingSpaces()
                        .withQuote('"')); // 明确指定引号字符

                List<CSVRecord> records = parser.getRecords();
                if (records.isEmpty()) {
                    System.err.println("空记录，跳过: " + line);
                    return;
                }

                CSVRecord record = records.get(0);

                // 验证记录是否包含足够的字段
                if (record.size() <= Math.max(SEASON_INDEX, Math.max(HOLIDAY_INDEX, CATEGORY_INDEX))) {
                    System.err.println("字段不足，跳过: " + line);
                    return;
                }

                // 提取季节
                String season = record.get(SEASON_INDEX).trim();
                if (season.isEmpty()) {
                    season = "Unknown"; // 处理空值
                }

                // 提取节假日信息（标准化为 Yes/No）
                String holiday = record.get(HOLIDAY_INDEX).trim();

                // 类别清洗（适配 [xxx, yyy] 格式）
                String categoryRaw = record.get(CATEGORY_INDEX);
                // 移除前后方括号和单引号，处理可能的引号转义
                String cleanedCategories = categoryRaw
                        .replaceAll("^\\[|\\]$|^'|'$", "")
                        .replaceAll("\\\\'", "'") // 处理可能的转义单引号
                        .trim();

                // 分割多品类
                String[] categories = cleanedCategories.split("\\s*,\\s*");

                for (String category : categories) {
                    if (!category.isEmpty()) {
                        // 构建输出 key，格式：季节,节假日,品类
                        outputKey.set(season + "," + holiday + "," + category);
                        context.write(outputKey, one);
                    }
                }

            } catch (Exception e) {
                System.err.println("未知错误: " + e.getMessage() + "，行: " + line);
            }
        }
    }

    // 季节性趋势分析 Reducer，统计每个（季节,节假日,品类）组合的销量
    public static class SeasonalTrendReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        private final IntWritable result = new IntWritable();

        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context)
                throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) {
                sum += val.get();
            }
            result.set(sum);
            context.write(key, result);
        }
    }

    // 区域市场分析 Mapper：提取 城市、子类别 维度，跳过第一行标题
    public static class RegionalMarketMapper extends Mapper<LongWritable, Text, Text, Text> {
        private final Text outputKey = new Text();
        private final Text outputValue = new Text();
        private boolean isFirstLine = true; // 标记是否为第一行

        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            String line = value.toString().trim();
            if (line.isEmpty()) return; // 跳过空行

            // 跳过第一行标题
            if (isFirstLine) {
                isFirstLine = false;
                System.out.println("跳过标题行: " + line);
                return;
            }

            String[] fields = line.split(",");
            // 确保字段足够（根据实际CSV列数调整，这里假设索引10是City，2是SubCategory）
            if (fields.length < 11) { // 至少需要11个字段才能访问索引10
                System.err.println("字段不足，跳过行: " + line);
                return;
            }

            String city = fields[10].trim();
            String subCategory = fields[2].trim();   // 子类别（SubCategory列，索引2）

            // 输出格式：<城市, 子类别>
            outputKey.set(city);
            outputValue.set(subCategory);
            context.write(outputKey, outputValue);
        }
    }

    // 区域市场分析 Reducer：统计各城市的热门子类别及销量
    public static class RegionalMarketReducer extends Reducer<Text, Text, Text, Text> {
        @Override
        protected void reduce(Text city, Iterable<Text> subCategories, Context context)
                throws IOException, InterruptedException {
            // 统计每个子类别在当前城市的销量
            Map<String, Integer> categoryCount = new HashMap<>();
            for (Text subCategory : subCategories) {
                String category = subCategory.toString();
                categoryCount.put(category, categoryCount.getOrDefault(category, 0) + 1);
            }

            // 按销量排序（降序）
            List<Map.Entry<String, Integer>> sortedCategories = new ArrayList<>(categoryCount.entrySet());
            sortedCategories.sort(Map.Entry.<String, Integer>comparingByValue().reversed());

            // 构造输出结果（取前3个热门子类别，可根据需要调整）
            StringBuilder result = new StringBuilder();
            int count = 0;
            for (Map.Entry<String, Integer> entry : sortedCategories) {
                if (count >= 3) break; // 只取前3个
                result.append(entry.getKey()).append(":").append(entry.getValue()).append(";");
                count++;
            }

            // 输出格式：<城市, 子类别1:销量;子类别2:销量;子类别3:销量>
            context.write(city, new Text(result.toString()));
        }
    }

    // 用户留存分析 Mapper：提取 用户ID、购买日期、订单金额 维度，跳过第一行标题
    public static class UserRetentionMapper extends Mapper<LongWritable, Text, Text, Text> {
        private final Text outputKey = new Text();
        private final Text outputValue = new Text();
        private boolean isFirstLine = true;

        // 正则表达式：匹配逗号但排除引号内的逗号
        private static final Pattern CSV_SPLIT_PATTERN =
                Pattern.compile(",(?=(?:[^\"]*\"[^\"]*\")*[^\"]*$)");

        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            String line = value.toString().trim();
            if (line.isEmpty()) return;

            // 跳过第一行标题
            if (isFirstLine) {
                isFirstLine = false;
                System.out.println("跳过标题行: " + line);
                return;
            }

            try {
                // 使用正则表达式分割CSV行，忽略引号内的逗号
                String[] fields = CSV_SPLIT_PATTERN.split(line);

                // 确保字段数量足够
                if (fields.length < 8) {
                    System.err.println("字段不足，跳过行: " + line);
                    return;
                }

                // 手动去除字段两端的引号
                for (int i = 0; i < fields.length; i++) {
                    fields[i] = fields[i].trim();
                    if (fields[i].startsWith("\"") && fields[i].endsWith("\"")) {
                        fields[i] = fields[i].substring(1, fields[i].length() - 1);
                    }
                }

                String userId = fields[0];
                String customerSegment = fields[6];
                double avgOrderValue = Double.parseDouble(fields[7]);

                // 输出格式：<用户类型, 平均订单价值>
                outputKey.set(customerSegment);
                outputValue.set(String.valueOf(avgOrderValue));
                context.write(outputKey, outputValue);
            } catch (NumberFormatException e) {
                System.err.println("解析数值错误，跳过行: " + line + " 错误: " + e.getMessage());
            } catch (Exception e) {
                System.err.println("解析错误，跳过行: " + line + " 错误: " + e.getMessage());
            }
        }
    }

    // 用户留存分析 Reducer
    public static class UserRetentionReducer extends Reducer<Text, Text, Text, Text> {
        private final Text result = new Text();

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            int count = 0;
            double totalOrderValue = 0;

            // 计算每个用户类型的总订单价值和订单数量
            for (Text value : values) {
                try {
                    double orderValue = Double.parseDouble(value.toString());
                    totalOrderValue += orderValue;
                    count++;
                } catch (NumberFormatException e) {
                    System.err.println("解析订单价值错误: " + e.getMessage());
                }
            }

            // 计算平均订单价值
            double averageOrderValue = count > 0 ? totalOrderValue / count : 0;

            // 输出结果：用户类型 -> 订单数量,总价值,平均价值
            result.set(String.format("订单数量: %d, 总价值: %.2f, 平均价值: %.2f",
                    count, totalOrderValue, averageOrderValue));

            context.write(key, result);
        }
    }
}